package cn.me.implement.threadpool.v3_0;

import cn.me.implement.threadpool.v3_0.reject.RejectPolicy;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * <h3>线程池 3.0 版本</h3>
 * 它实现的功能如下：
 * <ul>
 *     <li>在无法执行任务的情况下提供了拒绝策略</li>
 *     <li>
 *         提供给使用者一个构造方法，指定六个参数：
 *         <ul>
 *             <li><strong>核心线程数量</strong></li>
 *             <li><strong>最大线程数量</strong></li>
 *             <li><strong>临时线程在等待领取任务时最大阻塞时间</strong></li>
 *             <li><strong>阻塞时间的单位</strong></li>
 *             <li><strong>使用者指定的阻塞队列，充当存放任务的任务队列</strong></li>
 *             <li><strong>拒绝策略</strong></li>
 *         </ul>
 *     </li>
 *     <li>使用锁来保证 {@link #threadPool} 相关操作的 <strong>互斥性</strong></li>
 *     <li>
 *         将线程分为两类：
 *         <ul>
 *             <li><strong>核心线程</strong>：永不停止</li>
 *             <li><strong>临时线程</strong>：在一段时间内没有获取任务的情况下停止</li>
 *         </ul>
 *     </li>
 *     <li>使用一个 <strong>{@link Set}</strong> 存放所有线程</li>
 *     <li><strong>在新建线程时交给线程一个任务</strong></li>
 *     <li><strong>线程在提交任务时才创建</strong></li>
 *     <li><strong>{@link #submit(Runnable task)} 通过任务队列提交任务，线程通过任务队列领取任务</strong></li>
 * </ul>
 */
public class ThreadPool3_0 {

    /**
     * 线程池中核心线程的最大数量
     */
    private final int corePoolSize;

    /**
     * 线程池中线程的最大数量
     */
    private final int maxPoolSize;

    /**
     * 临时线程阻塞的最长时间（单位：ns），超过这个时间还没有领取到任务就直接退出
     */
    private final long keepAliveTime;

    /**
     * 任务队列
     */
    private final BlockingQueue<Runnable> taskQueue;

    /**
     * 拒绝策略，用于在无法执行任务的时候拒绝任务
     */
    private final RejectPolicy rejectPolicy;

    /**
     * 构造一个线程池
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     * @param keepAliveTime 临时线程阻塞的最长时间
     * @param unit 时间的单位
     * @param taskQueue 任务队列
     * @param rejectPolicy 拒绝策略
     */
    public ThreadPool3_0(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> taskQueue, RejectPolicy rejectPolicy) {
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.taskQueue = taskQueue;
        this.rejectPolicy = rejectPolicy;
    }

    /**
     * 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高
     */
    private final Set<Worker> threadPool = new HashSet<>();

    /**
     * 线程池的管程
     * <p>
     * 用于保证 <strong>获取线程池大小</strong>、<strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性
     */
    private final Object threadPoolMonitor = new Object();

    /**
     * <h3>核心线程执行的任务</h3>
     * {@link #getTask()} 方法会一直阻塞，直到有新任务
     */
    public final class CoreWorker extends Worker {

        public CoreWorker(Runnable initialTask, Set<Worker> threadPool) {
            super(initialTask, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        protected void onWorkerExit() {
            // 在目前的代码中，发现核心线程并不会退出，所以这个方法先不实现
        }
    }

    /**
     * <h3>临时线程执行的任务</h3>
     * {@link #getTask()} 方法会在阻塞一定时间后如果还没有任务，则会返回 {@code null}
     */
    public final class TempWorker extends Worker {

        public TempWorker(Runnable initialTask, Set<Worker> threadPool) {
            super(initialTask, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        protected void onWorkerExit() {
            removeWorkerFromThreadPool(this);
        }
    }

    /**
     * 提交任务
     *
     * @param task 待执行的任务
     */
    public void submit(Runnable task) {
        // 如果 线程数量 小于 最大核心线程数量，则新建一个 核心线程 执行任务，然后直接返回
        synchronized (threadPoolMonitor) {
            if (threadPool.size() < corePoolSize) {
                CoreWorker coreWorker = new CoreWorker(task, threadPool);
                coreWorker.start();
                return;
            }
        }

        // 如果能够放到任务队列中，则直接返回
        if (taskQueue.offer(task)) {
            return;
        }

        // 如果 线程数量 小于 最大线程数量，则新建一个 临时线程 执行任务
        synchronized (threadPoolMonitor) {
            if (threadPool.size() < maxPoolSize) {
                TempWorker tempWorker = new TempWorker(task, threadPool);
                tempWorker.start();
                return;
            }
        }

        // 线程数量到达最大线程数量，任务队列已满，执行拒绝策略
        rejectPolicy.reject(this, task);
    }

    /**
     * 获取当前线程池中的线程数量
     *
     * @return 当前线程池中的线程数量
     */
    public int getCurrPoolSize() {
        synchronized (threadPoolMonitor) {
            return threadPool.size();
        }
    }

    /**
     * 丢弃任务队列 {@link #taskQueue} 中的最旧的任务（队头任务）
     *
     * @return 任务队列中的最旧的任务（队头任务）
     */
    public Runnable discardOldestTask() {
        return taskQueue.poll();
    }

    /**
     * 从 {@link #threadPool} 中移除指定的 {@link Worker} 对象
     *
     * @param worker 待移除的 {@link Worker} 对象
     */
    private void removeWorkerFromThreadPool(Worker worker) {
        synchronized (threadPoolMonitor) {
            threadPool.remove(worker);
        }
    }
}
